package com.fudan.run.trans;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import org.apache.spark.api.java.JavaRDD;

import com.fudan.cfg.trans.FlatRpcTransDef;
import com.fudan.run.JobRunner;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.TransAdapter;
import com.lobstar.xrpc.client.main.MainClient;
import com.lobstar.xrpc.tools.Fields;

@TransAdapter("flat_rpc")
public class FlatRpcTrans extends TransHandler<FlatRpcTransDef>{

	@Override
	public void trans(JobRunner jobRunner,FlatRpcTransDef transNode, VariableRddMap variableMap) {
		String target = transNode.getTarget();
		JavaRDD<Map<String, Object>> targetRdd = variableMap.getRdd(target);
		JavaRDD<Map<String, Object>> after = targetRdd.mapPartitions(l->{
			List<Map<String,Object>> list = new ArrayList<>();
			MainClient client = MainClient.build(UUID.randomUUID().toString(), transNode.getIp(), transNode.getPort(), null).blockStart();
			while(l.hasNext()) {
				Map<String, Object> map = l.next();
				List<Map<String,Object>> ml = client.asyncCall(transNode.getMethod(), Fields.build().field("map", map), List.class);
				list.addAll(ml);
			}
			return list.iterator();
		});
		variableMap.addRdd(transNode.getName(), after);
	}

}
